package com.bj58.mis.apm.server.kylinreceive.service.impl;

import com.bj58.mis.apm.server.core.util.JsonUtil;
import com.bj58.mis.apm.server.core.util.MD5;
import com.bj58.mis.apm.server.kylinreceive.dto.KylinData;
import com.bj58.mis.apm.server.kylinreceive.model.*;
import com.bj58.mis.apm.server.kylinreceive.repository.*;
import com.bj58.mis.apm.server.kylinreceive.service.KylinDataService;
import com.bj58.mis.apm.server.kylinreceive.service.WarnService;
import org.apache.commons.beanutils.BeanUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.transaction.Transactional;
import java.lang.reflect.InvocationTargetException;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

@Service
@Transactional
public class KylinDataServiceImpl implements KylinDataService {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private KylinProjectRepository projectRepository;
    @Resource
    private KylinCubedescRepository cubedescRepository;
    @Resource
    private KylinCubedescRowkeyRepository rowkeyRepository;
    @Resource
    private KylinCubeRepository cubeRepository;
    @Resource
    private KylinTableRepository tableRepository;
    @Resource
    private KylinModelRepository modelRepository;
    @Resource
    private KylinJobRepository jobRepository;
    @Resource
    private KylinJobStepRepository jobStepRepository;
    @Resource
    private KylinJobStepOutputRepository jobStepOutputRepository;
    @Resource
    private KylinSegmentRepository segmentRepository;
    @Resource
    private KylinCubeStatisticRepository statisticRepository;
    @Resource
    private StorageRepository storageRepository;

    @Resource
    private WarnService warnService;

    private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

    @Override
    public void receive(KylinData kylinData) {
        logger.debug("处理kylin数据接收入库");
        saveKylinData(kylinData);
        logger.debug("处理kylin数据接收入库完成");
    }

    private void saveKylinData(KylinData kylinData){
        Long lastModified = kylinData.getLastModified();
        if(lastModified == null){
            lastModified = 0L;
        }

        //处理项目更新数据
        saveProjects(kylinData.getProjects(), lastModified);
        //处理更新的table
        saveTables(kylinData.getTables());
        //处理更新的model
        saveModels(kylinData.getModels());
        //处理更新的cubedesc
        saveCubeDesc(kylinData.getCubeDescs());
        //处理更新的cube
        saveCube(kylinData.getCubes());
        //处理更新的job
        saveJobs(kylinData.getJobs(), kylinData.getJobOutputs());
        //保存cuboids信息
        saveCuboids(kylinData.getCuboids());

        warnService.checkBuild(kylinData.getJobs());
    }

    /**
     * 处理项目更新数据
     * @param projects
     * @param lastModified
     */
    private void saveProjects(List<Map<String, Object>> projects, Long lastModified){
        //再用的project
        List<String> projectIds = new ArrayList<>();
        //cube与project的映射关系
        Map<String, String> cubeToProject = new HashMap<>();
        //table与project的映射关系
        Map<String, String> tableToProject = new HashMap<>();
        //model与project的映射关系
        Map<String, String> modelToProject = new HashMap<>();

        logger.debug("=======project================== size:"+projects.size());
        int modified = 0;
        for(Map<String, Object> project : projects){
            projectIds.add((String)project.get("uuid"));

            if(((Double)project.get("last_modified")).longValue() <= lastModified){
                continue;
            }
            modified ++;
            KylinProject kylinProject = saveProject(project);

            //处理项目关联model的数据，不存在的标记为删除，并记录删除时间
            List<String> modelNames = (List<String>)project.get("models");
            List<KylinModel> modelList = modelRepository.queryByProjectIdAndIsDelete(kylinProject.getId(), false);
            modelList.forEach( m -> {
                if(!modelNames.contains(m.getName())){
                    m.setDelete(true);
                    m.setDeleteTime(new Date());
                    modelRepository.save(m);
                }
            });
            //处理项目关联table的数据，不存在的标记为删除，并记录删除时间
            List<String> tableNmaes = (List<String>)project.get("tables");
            List<KylinTable> tableList = tableRepository.queryByProjectIdAndIsDelete(kylinProject.getId(), false);
            tableList.forEach( t -> {
                if(!tableNmaes.contains(t.getName())){
                    t.setDelete(true);
                    t.setDeleteTime(new Date());
                    tableRepository.save(t);
                }
            });

            //处理不在项目下的cube数据，标记为删除，并记录删除时间
            List<String> cubeNames = new ArrayList<>();
            List<Map<String, Object>> realizations = (List<Map<String, Object>>)project.get("realizations");
            for(Map<String, Object> map : realizations){
                if("CUBE".equals(map.get("type"))){
                    cubeNames.add((String)map.get("realization"));
                }
            }
            List<KylinCube> cubeList = cubeRepository.queryByProjectIdAndIsDelete(kylinProject.getId(), false);
            cubeList.forEach( cube -> {
                if(!cubeNames.contains(cube.getNameEn())){
                    cube.setDelete(true);
                    cube.setDeleteTime(new Date());
                    cubeRepository.save(cube);
                    List<KylinSegment> segments = segmentRepository.queryByCubeIdAndIsDelete(cube.getId(), false);
                    segments.forEach( s -> {
                        s.setDelete(true);
                        s.setDeleteTime(new Date());
                        segmentRepository.save(s);
                    });
                }
            });
        }
        logger.debug("=======project modified================== size:"+modified);

        //处理不在project数据列表中的数据，标记为删除，并记录删除时间
        List<KylinProject> projectList = projectRepository.queryByIsDelete(false);
        projectList.forEach( p -> {
            if(!projectIds.contains(p.getId())){
                p.setDelete(true);
                p.setDeleteTime(new Date());
                projectRepository.save(p);
            }
        });

    }

    /**
     * 保存单个project
     * @param project
     */
    protected KylinProject saveProject(Map<String, Object> project){

        KylinProject kylinProject = new KylinProject();
        populate(kylinProject, project);

        kylinProject.setId(getString(project, "uuid"));
        kylinProject.setNameEn(kylinProject.getName());
        kylinProject.setLastModified(getDate(project, "last_modified"));
        kylinProject.setCreateTimeUtc(getDate(project, "create_time_utc"));
        kylinProject.setTables(getJson(project, "tables"));
        kylinProject.setModels(getJson(project, "models"));
        kylinProject.setRealizations(getJson(project, "realizations"));
        kylinProject.setDelete(false);
        projectRepository.save(kylinProject);

        return kylinProject;
    }

    /**
     * 保存更新的cubedesc
     * @param cubeDescs
     */
    protected void saveCubeDesc(List<Map<String, Object>> cubeDescs){
        logger.debug("=======cubedesc================== size:"+cubeDescs.size());
        Map<String, String> tempMap = new HashMap<>();

        for(Map<String, Object> cubedesc : cubeDescs){
            KylinCubedesc kylinCubedesc = new KylinCubedesc();
            populate(kylinCubedesc, cubedesc);

            kylinCubedesc.setId(getString(cubedesc, "uuid"));
            kylinCubedesc.setLastModified(getDate(cubedesc, "last_modified"));
            kylinCubedesc.setIsDraft(getBoolean(cubedesc, "is_draft"));

            kylinCubedesc.setModelName(getString(cubedesc, "model_name"));
            kylinCubedesc.setNullString(getString(cubedesc, "null_string"));
            kylinCubedesc.setDimensions(getJson(cubedesc, "dimensions"));
            kylinCubedesc.setMeasures(getJson(cubedesc, "measures"));
            kylinCubedesc.setRowkey(getJson(cubedesc, "rowkey"));
            kylinCubedesc.setHbaseMapping(getJson(cubedesc, "hbase_mapping"));
            kylinCubedesc.setAggregationGroups(getJson(cubedesc, "aggregation_groups"));
            kylinCubedesc.setCuboidBlackList(getJson(cubedesc, "cuboid_black_list"));
            kylinCubedesc.setParentForward(getLong(cubedesc,"parent_forward"));

            kylinCubedesc.setPartitionDateStart(getDate(cubedesc, "partition_date_start"));
            kylinCubedesc.setVolatileRange(getLong(cubedesc,"volatile_range"));
            kylinCubedesc.setRetentionRange(getLong(cubedesc,"retention_range"));
            kylinCubedesc.setEngineType(getLong(cubedesc,"engine_type"));
            kylinCubedesc.setStorageType(getLong(cubedesc,"storage_type"));
            Collection ranges = getCollection(cubedesc, "auto_merge_time_ranges");

            if(ranges != null && ranges.size() > 0) {
                List<Long> autoRanges = new ArrayList<>();
                for (Object obj : ranges) {
                    autoRanges.add(((Number) obj).longValue());
                }
                kylinCubedesc.setAutoMergeTimeRanges(JsonUtil.toJson(autoRanges));
            }
            Map<String, Object> rowkey = (Map<String, Object>)cubedesc.get("rowkey");
            List<Map<String, Object>> columns = (List<Map<String, Object>>)rowkey.get("rowkey_columns");
            kylinCubedesc.setFullMark(fullMack(columns.size()));

            cubedescRepository.save(kylinCubedesc);

            rowkeyRepository.deleteByCubedescId(kylinCubedesc.getId());

            for(int i = 0; i < columns.size(); i++){
                Map<String, Object> column = columns.get(i);
                KylinCubedescRowkey cubeRowkey = new KylinCubedescRowkey();
                this.populate(cubeRowkey, column);

                String col = getString(column, "column");
                String[] cols = col.split("[.]");
                cubeRowkey.setTableName(cols[0]);
                cubeRowkey.setColumnName(cols[1]);
                cubeRowkey.setIndexNumber(i+1L);
                cubeRowkey.setCubedescId(kylinCubedesc.getId());
                cubeRowkey.setShardBy(getBoolean(column, "isShardBy"));

                String dataType = tempMap.get(col);
                if(dataType == null){
                    KylinTable table = tableRepository.getKylinTableByProjectIdAndNameAndIsDelete(kylinCubedesc.getProjectId(), cubeRowkey.getTableName(), false);
                    if(table != null){
                        List<Map<String, Object>> columnList = JsonUtil.fromArrayJson(table.getColumns());
                        columnList.forEach(c -> {
                            tempMap.put(table.getName() + "." + c.get("name"), (String) c.get("datatype"));
                        });
                    }
                    dataType = tempMap.get(col);
                }
                cubeRowkey.setDataType(dataType);

                rowkeyRepository.save(cubeRowkey);
            }
        }
    }

    /**
     * 保存更新的cube
     * @param cubes
     */
    protected void saveCube(List<Map<String, Object>> cubes){
        logger.debug("=======cube================== size:"+cubes.size());
        updateStorage();
        Date newDate = getNowDate();
        for(Map<String, Object> cube : cubes) {
            KylinCube kylinCube = new KylinCube();
            populate(kylinCube, cube);

            kylinCube.setId(getString(cube, "uuid"));
            kylinCube.setNameEn(kylinCube.getName());
            kylinCube.setLastModified(getDate(cube, "last_modified"));
            kylinCube.setDisplayName(getString(cube, "display_name"));
            kylinCube.setCreateTimeUtc(getDate(cube, "create_time_utc"));
            kylinCube.setCuboidBytes(getString(cube, "cuboid_bytes"));
            kylinCube.setCuboidBytesRecommend(getString(cube, "cuboid_bytes_recommend"));
            kylinCube.setCuboidLastOptimized(getLong(cube, "cuboid_last_optimized"));
            kylinCube.setDelete(false);

            cubeRepository.save(kylinCube);

            List<String> active = new ArrayList<>();
            List<Map<String, Object>> segments = (List<Map<String, Object>>) cube.get("segments");
            List<KylinSegment> segmentListForWarn = new ArrayList<>();
            for (Map<String, Object> segment : segments) {
                KylinSegment kylinSegment = new KylinSegment();
                this.populate(kylinSegment, segment);

                kylinSegment.setId(getString(segment, "uuid"));
                kylinSegment.setDateRangeStart(getDate(segment, "date_range_start"));
                kylinSegment.setDateRangeEnd(getDate(segment, "date_range_end"));
                kylinSegment.setSourceOffsetStart(getLong(segment, "source_offset_start"));
                kylinSegment.setSourceOffsetEnd(getLong(segment, "source_offset_end"));
                kylinSegment.setSize(getLong(segment, "size_kb") * 1000);
                kylinSegment.setInputRecords(getLong(segment, "input_records"));
                kylinSegment.setInputRecordsSize(getLong(segment, "input_records_size"));
                kylinSegment.setLastBuildTime(getDate(segment, "last_build_time"));
                kylinSegment.setLastBuildJobId(getString(segment, "last_build_job_id"));
                kylinSegment.setCreateTimeUtc(getDate(segment, "create_time_utc"));
                kylinSegment.setDictionaries(getJson(segment, "dictionaries"));
                kylinSegment.setRowkeyStats(getJson(segment, "rowkey_stats"));
                if (kylinSegment.getSize() != null && kylinSegment.getInputRecordsSize() != null && kylinSegment.getSize() > 0 && kylinSegment.getInputRecordsSize() > 0) {
                    kylinSegment.setSwell(kylinSegment.getSize().doubleValue() / kylinSegment.getInputRecordsSize());
                }
                kylinSegment.setCubeId(kylinCube.getId());
                kylinSegment.setDelete(false);
                active.add(kylinSegment.getId());

                segmentRepository.save(kylinSegment);
                segmentListForWarn.add(kylinSegment);
            }
            warnService.checkSwell(segmentListForWarn, kylinCube);

            //处理不在cube中办好的segment数据，即segment已过期
            List<KylinSegment> segmentList = segmentRepository.queryByCubeIdAndIsDelete(kylinCube.getId(), false);
            segmentList.forEach(s -> {
                if (!active.contains(s.getId())) {
                    s.setDelete(true);
                    s.setDeleteTime(new Date());
                    segmentRepository.save(s);
                }
            });

            //根据更新的cube进行统计
            Map<String, Object> totalSize = segmentRepository.queryCubeTotalSize(kylinCube.getId());
            if(totalSize != null && totalSize.get("hbaseDataSize") != null && totalSize.get("metaDataSize") != null) {
                KylinStorage storage = storageRepository.getByCubeIdAndRecordDate(kylinCube.getId(), newDate);
                if (storage == null) {
                    storage = new KylinStorage();
                    storage.setCubeId(kylinCube.getId());
                    storage.setRecordDate(newDate);
                }
                storage.setHbaseDataSize(((BigDecimal) totalSize.get("hbaseDataSize")).doubleValue());
                storage.setMetaDataSize(((BigDecimal) totalSize.get("metaDataSize")).doubleValue());
                storageRepository.save(storage);
            }
        }
    }

    /**
     * 保存更新的table数据
     * @param tables
     */
    @Transactional
    protected void saveTables(List<Map<String, Object>> tables){
        logger.debug("=======tables================== size:"+tables.size());
        for(Map<String, Object> table : tables){
            KylinTable kylinTable = new KylinTable();
            this.populate(kylinTable, table);

            kylinTable.setId(getString(table,"uuid"));
            kylinTable.setLastModified(getDate(table, "last_modified"));
            kylinTable.setColumns(getJson(table, "columns"));
            kylinTable.setSourceType(getLong(table, "source_type"));
            kylinTable.setTableType(getString(table, "table_type"));
            kylinTable.setDelete(false);

            tableRepository.save(kylinTable);
        }
    }

    /**
     * 保存更新的model数据
     * @param models
     */
    protected void saveModels(List<Map<String, Object>> models){
        logger.debug("=======models================== size:"+models.size());
        for(Map<String, Object> model : models){
            KylinModel kylinModel = new KylinModel();
            this.populate(kylinModel, model);

            kylinModel.setId(getString(model, "uuid"));
            kylinModel.setLastModified(getDate(model, "last_modified"));
            kylinModel.setDraft(getBoolean(model, "is_draft"));
            kylinModel.setFactTable(getString(model, "fact_table"));
            kylinModel.setLookups(getJson(model, "lookups"));
            kylinModel.setDimensions(getJson(model, "dimensions"));
            kylinModel.setMetrics(getJson(model, "metrics"));
            kylinModel.setFilterCondition(getString(model, "filter_condition"));
            kylinModel.setPartitionDesc(getJson(model, "partition_desc"));
            kylinModel.setDelete(false);

            modelRepository.save(kylinModel);
        }
    }

    /**
     * 处理job和分步输出
     * @param jobs
     * @param jobOutputs
     */
    private void saveJobs(List<Map<String,Object>> jobs, List<Map<String,Object>> jobOutputs) {
        logger.debug("=======jobs================== size:"+jobs.size());
        jobs.forEach(job -> {
            KylinJob kylinJob = new KylinJob();
            this.populate(kylinJob, job);

            kylinJob.setId(getString(job, "uuid"));
            kylinJob.setLastModified(getDate(job, "last_modified"));
            kylinJob.setSegmentId(getString(job, "related_segment"));
            kylinJob.setExecStartTime(getDate(job, "exec_start_time"));
            kylinJob.setExecEndTime(getDate(job, "exec_end_time"));
            kylinJob.setExecInterruptTime(getLong(job, "exec_interrupt_time"));
            kylinJob.setMrWaiting(getLong(job, "mr_waiting"));
            kylinJob.setJobStatus(getString(job, "job_status"));

            List<Map<String, Object>> steps = (List<Map<String, Object>>)job.get("steps");

            if(steps.size() > 0){
                if(kylinJob.getExecStartTime() == null){
                    kylinJob.setExecStartTime(getDate(steps.get(0), "exec_start_time"));
                }
                if(kylinJob.getExecEndTime() == null) {
                    for (int i = steps.size() - 1; i >= 0; i--) {
                        kylinJob.setExecEndTime(getDate(steps.get(i), "exec_end_time"));
                    }
                }
            }
            jobRepository.save(kylinJob);

            steps.forEach( step -> {
                KylinJobStep kylinJobStep = new KylinJobStep();
                this.populate(kylinJobStep, step);

                kylinJobStep.setInterruptCmd1(getString(step, "interruptCmd"));
                kylinJobStep.setInterruptCmd2(getString(step, "interrupt_cmd"));
                kylinJobStep.setSequenceId(getLong(step, "sequence_id"));
                kylinJobStep.setExecStartTime(getDate(step, "exec_start_time"));
                kylinJobStep.setExecEndTime(getDate(step, "exec_end_time"));
                if(kylinJobStep.getExecEndTime() != null && kylinJobStep.getExecEndTime() != null) {
                    kylinJobStep.setDuration(kylinJobStep.getExecEndTime().getTime() - kylinJobStep.getExecStartTime().getTime());
                }
                kylinJobStep.setExecWaitTime(getLong(step, "exec_wait_time"));
                kylinJobStep.setStepStatus(getString(step, "step_status"));
                kylinJobStep.setCmdType(getString(step, "cmd_type"));
                kylinJobStep.setInfo(getJson(step, "info"));
                kylinJobStep.setRunAsync(getBoolean(step, "run_async"));
                kylinJobStep.setJobId(kylinJob.getId());

                jobStepRepository.save(kylinJobStep);
            });

        });

        logger.debug("=======jobOutputs================== size:"+jobOutputs.size());
        jobOutputs.forEach( output -> {
            KylinJobStepOutput op = new KylinJobStepOutput();
            this.populate(op, output);

            op.setCmdOutput(getString(output, "cmd_output"));

            jobStepOutputRepository.save(op);
        });

    }

    private void saveCuboids(List<Map<String,Object>> cuboids) {
        cuboids.forEach( statistic -> {
            KylinCubeStatistic kylinCubeStatistic = new KylinCubeStatistic();
            this.populate(kylinCubeStatistic, statistic);

            kylinCubeStatistic.setLastModified(getDate(statistic, "last_modified"));
            kylinCubeStatistic.setId(MD5.encode(kylinCubeStatistic.getCubeId(), "" + kylinCubeStatistic.getLastModified().getTime()));

            Map<String, Object> root = (Map<String, Object>)statistic.get("root");
            kylinCubeStatistic.setCuboids(getCuboidCount(root));
            kylinCubeStatistic.setCuboidText(JsonUtil.toJson(root));
            statisticRepository.save(kylinCubeStatistic);

        });
    }

    private long getCuboidCount(Map<String, Object> root){
        long count = 1L;
        List<Map<String, Object>> children = (List<Map<String, Object>>)root.get("children");
        if(children != null){
            for(Map<String, Object> c : children){
                count += getCuboidCount(c);
            }
        }
        return count;
    }

    /**
     * 将Map数据转换成Object
     * @param t
     * @param map
     * @param <T>
     */
    private <T> void populate(T t, Map map){
        try {
            BeanUtils.populate(t, map);
        } catch (IllegalAccessException e) {
            e.printStackTrace();
        } catch (InvocationTargetException e) {
            e.printStackTrace();
        }
    }

    /**
     * 将Map中的集合数据和对象数据转换成json字符串返回
     * @param map
     * @param key
     * @return
     */
    private String getJson(Map<String, Object> map, String key){
        Object obj = map.get(key);
        if(obj instanceof List){
            return ((List)obj).size() == 0 ? null : JsonUtil.toJson(obj);
        }else{
            return JsonUtil.toJson(obj);
        }
    }

    /**
     * 将固定key的值转换成Date类型
     * @param map
     * @param key
     * @return
     */
    private Date getDate(Map<String, Object> map, String key){
        Long value = ((Double)map.get(key)).longValue();
        if(value > 0 && value < Long.MAX_VALUE){
            return new Date(value);
        }
        return null;
    }

    /**
     * 读取Long数据
     * @param map
     * @param key
     * @return
     */
    private Long getLong(Map<String, Object> map, String key){
        return ((Number)map.get(key)).longValue();
    }

    private String getString(Map<String, Object> map, String key) {
        return (String) map.get(key);
    }

    private Collection getCollection(Map<String, Object> map, String key) {
        return (Collection)map.get(key);
    }

    private Boolean getBoolean(Map<String, Object> map, String key) {
        return (Boolean)map.get(key);
    }

    /**
     * 计算fullMack
     * @param length
     * @return
     */
    public Long fullMack(int length){
        long full_mark = 0L;
        for(int i = 0; i < length; i++){
            full_mark |= 1L << i;
        }
        return full_mark;
    }

    private void updateStorage(){

        Date nowDate = getNowDate();

        Date lastRecordDate = storageRepository.queryLastRecordDate();

        //没有数据时处理第一条
        if (lastRecordDate == null) {
            updateNowStorage();
        } else if (nowDate.getTime() > lastRecordDate.getTime()) {
            Calendar cal = Calendar.getInstance();
            cal.setTime(lastRecordDate);
            Long now = nowDate.getTime();
            cal.add(Calendar.DATE, 1);
            while (cal.getTimeInMillis() < now) {
                storageRepository.updateRecords(cal.getTime());
                cal.add(Calendar.DATE, 1);
            }
            updateNowStorage();
        }
    }

    @Override
    public void updateNowStorage(){
        Date newDate = getNowDate();
        List<Map<String, Object>> list = segmentRepository.queryAllCubeTotalSize();
        for (Map<String, Object> m : list) {
            KylinStorage storage = storageRepository.getByCubeIdAndRecordDate((String) m.get("cubeId"), newDate);
            if (storage == null) {
                storage = new KylinStorage();
                storage.setCubeId((String) m.get("cubeId"));
                storage.setRecordDate(newDate);
            }
            storage.setHbaseDataSize(((BigDecimal) m.get("hbaseDataSize")).doubleValue());
            storage.setMetaDataSize(((BigDecimal) m.get("metaDataSize")).doubleValue());
            storageRepository.save(storage);
        }
    }

    /**
     * 得到当前日期
     * @return
     */
    private Date getNowDate(){
        try {
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(sdf.parse(sdf.format(new Date())));
            return calendar.getTime();
        }catch (ParseException e){
            throw new RuntimeException(e.getMessage());
        }
    }
}
